package com.cms.pms.aggregator;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.tomcat.util.codec.binary.StringUtils;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.impl.InfluxDBImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

import com.cms.pms.Aggregator;
import com.cms.pms.PMSContext;
import com.cms.pms.Store;
import com.cms.pms.Transaction;

import io.netty.util.internal.StringUtil;

public class SimpleAggregator implements Aggregator {

	private static String EXP_LOG = "log";
	private static String EXP_INFLUXDB = "influxdb";
	private static String EXP_MQ = "mq";

	Logger log = LoggerFactory.getLogger(SimpleAggregator.class);

	public Map<String, List<Transaction>> transMap = null;

	public Map<String, Indexes> indexMap = null;

	public Converter convertor = null;

	private String influxdbUrl = "http://localhost:8086";

	private String influxdbUser = "root";
	private String influxdbPwd = "root";
	private String influxdbDb = "example";

	private String export = EXP_LOG;

	private JmsTemplate jmstemplete;

	private String extraTags;

	public String getExtraTags() {
		return extraTags;
	}

	public void setExtraTags(String extraTags) {
		this.extraTags = extraTags;
	}

	public String getInfluxdbUrl() {
		return influxdbUrl;
	}

	public void setInfluxdbUrl(String influxdbUrl) {
		this.influxdbUrl = influxdbUrl;
	}

	public String getInfluxdbUser() {
		return influxdbUser;
	}

	public void setInfluxdbUser(String influxdbUser) {
		this.influxdbUser = influxdbUser;
	}

	public String getInfluxdbPwd() {
		return influxdbPwd;
	}

	public void setInfluxdbPwd(String influxdbPwd) {
		this.influxdbPwd = influxdbPwd;
	}

	public String getInfluxdbDb() {
		return influxdbDb;
	}

	public void setInfluxdbDb(String influxdbDb) {
		this.influxdbDb = influxdbDb;
	}

	public String getExport() {
		return export;
	}

	public void setExportType(String export) {
		this.export = export;
	}

	public Converter getConvertor() {
		return convertor;
	}

	public void setConvertor(Converter convertor) {
		this.convertor = convertor;
	}

	public void setMapping(Map map) {
		SimpleConverter converter = new SimpleConverter();
		converter.setMapping(map);
		this.convertor = converter;
	}

	public SimpleAggregator() {
		transMap = new HashMap<String, List<Transaction>>();
		indexMap = new HashMap<String, Indexes>();
	}

	public Map<String, Indexes> getIndexes() {
		return indexMap;
	}

	@Override
	public void handle(Object obj) {

		if (obj instanceof Transaction) {
			Transaction trans = (Transaction) obj;
			String name = trans.signature();
			if (convertor != null) {
				name = convertor.getMetric(name);
			}
			List<Transaction> list = transMap.get(name);
			if (list == null) {
				list = new ArrayList<Transaction>();
				transMap.put(name, list);
			}
			list.add(trans);
		}
	}

	@Override
	public void aggregate() {

		this.calculate();
		
		
		if (export.indexOf(EXP_LOG) > -1) {
			this.write2log();
		}

		if (export.indexOf(EXP_INFLUXDB) > -1) {
			this.send2Influxdb();
		}

		if (export.indexOf(EXP_MQ) > -1) {
			this.send2mq();
		}

	}

	@Override
	public void prepare() {
		this.clear();
	}

	private void clear() {
		for (List l : transMap.values()) {
			l.clear();
		}
		indexMap.clear();
	}

	private void calculate() {
		for (String name : transMap.keySet()) {
			List<Transaction> list = transMap.get(name);
			if (list == null || list.size() == 0)
				continue;
			Indexes index = new Indexes();
			indexMap.put(name, index);

			if (!org.springframework.util.StringUtils.isEmpty(extraTags))
				name = name + "," + extraTags;

			index.setName(name);

			int count = list.size();
			index.setCount(count);

			long sum = 0;
			long max = Long.MIN_VALUE;
			long min = Long.MAX_VALUE;

			HashSet<Thread> tset = new HashSet<Thread>();

			long d = 0;

			for (Transaction tr : list) {
				d = tr.getDuration();
				sum += d;
				max = max > d ? max : d;
				min = min < d ? min : d;

				tset.add(tr.getThread());
			}
			index.setAvg(sum / count);
			index.setMax(max);
			index.setMin(min);
			index.setThreadnum(tset.size());
		}
	}

	void send2Influxdb() {
		InfluxDB influxdb = InfluxDBFactory.connect(influxdbUrl, influxdbUser, influxdbDb);

		influxdb.createDatabase(influxdbDb);

		influxdb.enableBatch(1000, 100, TimeUnit.MILLISECONDS);

		String hostname = this.getHostname();

		for (Indexes i : indexMap.values()) {
			Point point = Point.measurement(i.getName()).tag("host", hostname).addField("count", i.getCount())
					.addField("max", i.getMax()).addField("min", i.getMin()).addField("avg", i.getAvg())
					.addField("threadnum", i.getThreadnum()).time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
					.build();
			influxdb.write(influxdbDb, "default", point);
		}
	}

	void write2log() {
		String hostname = this.getHostname();
		for (Indexes i : indexMap.values()) {
			log.info("PMS\t {},host={}\t count={},avg={},max={},min={},threadnum={}", i.getName(), hostname,
					i.getCount(), i.getAvg(), i.getMax(), i.getMin(), i.getThreadnum());

		}
	}

	void send2mq() {
		
		if (jmstemplete != null) {
			
			String hostname = this.getHostname();
			for (Indexes i : indexMap.values()) {
				final StringBuilder sb = new StringBuilder();
				sb.append(i.getName()).append(",host=").append(hostname);
				sb.append("\t").append("count=").append(i.getCount()).append(",avg=").append(i.getAvg()).append(",max=")
						.append(i.getMax()).append(",min=").append(i.getMin()).append(",threadnum=")
						.append(i.getThreadnum());
				jmstemplete.send(new MessageCreator() {
					@Override
					public Message createMessage(Session arg0) throws JMSException {
						TextMessage tmsg = arg0.createTextMessage();
						tmsg.setText(sb.toString());
						return tmsg;
					}
				});
			}
		}else
		{
			log.warn("The jmstemple is null");
		}
	}

	private String getHostname() {
		InetAddress address = null;
		try {
			address = InetAddress.getLocalHost();
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}
		return address.getHostName();
	}

	public void handle() {

		Map<Thread, Store> map = PMSContext.getAllStore();
		
		this.clear();
		
		for (Store s : map.values()) {
			while (true) {
				Transaction trans = (Transaction) s.take();
				if (trans == null)
					break;
				String name = trans.signature();
				List<Transaction> list = transMap.get(name);
				if (list == null) {
					list = new ArrayList<Transaction>();
					transMap.put(name, list);
				}
				list.add(trans);
			}
		}

		this.calculate();

		this.send2Influxdb();

	}

	public static void main(String[] args) {
		SimpleAggregator aggregator = new SimpleAggregator();
		Indexes index = new Indexes();
		index.setName("call");
		index.setCount(1000);
		index.setMax(100L);
		index.setMin(1L);
		index.setAvg(50L);
		Map<String, Indexes> map = aggregator.getIndexes();
		map.put("call", index);
		aggregator.send2Influxdb();
	}
}